package tails

import (
	"context"
	"encoding/json"
	"github.com/hpcloud/tail"
	"github.com/sirupsen/logrus"
	clientv3 "go.etcd.io/etcd/client/v3"
	"strings"
	"time"
	"tym.com/LogCollection/mq"
)

func TailTaskRun() {
	for _, task := range TailTask {
		go func() {
			for {
				select {
				case <-TailTaskContext.Done():
					logrus.Infof("tail: tail %s file is stop", task.Path)
					task.Tail.Cleanup()
					return
				case line, ok := <-task.Tail.Lines:
					if !ok {
						logrus.Warnf("tail file close reopen. filename: %s \n", task.Path)
						time.Sleep(time.Second)
						continue
					}
					if strings.EqualFold(line.Text, "\r") {
						continue
					}
					mq.SendKafkaMsgChan(task.Topic, line.Text)
				}
			}
		}()
	}
}

func TailTaskWatch(watchChan clientv3.WatchChan) {
	for {
		select {
		case wresp := <-watchChan:
			TailTaskCancel()
			for _, evt := range wresp.Events {
				if evt.Type == clientv3.EventTypeDelete {
					logrus.Infof("Watch: key %s stop!\n", evt.Kv.Key)
					return
				}
				err := json.Unmarshal(evt.Kv.Value, &TailTask)
				if err != nil {
					logrus.Error("json convert defeat!", evt.Kv.Value)
					return
				}
				//重新开启追踪
				TailTaskContext, TailTaskCancel = context.WithCancel(context.Background())
				for _, task := range TailTask {
					t, err := tail.TailFile(task.Path, tailConfig)
					task.Tail = t
					if err != nil {
						logrus.Error("tails:", err)
						return
					}
				}
				TailTaskRun()

			}
		}
	}
}
